package com.winit.cloudlink.message;

import com.google.common.collect.Maps;
import com.winit.cloudlink.common.Clearable;
import com.winit.cloudlink.common.Lifecycle;
import com.winit.cloudlink.common.exception.CloudlinkException;
import com.winit.cloudlink.config.Metadata;
import com.winit.cloudlink.message.handler.MessageHandler;
import com.winit.cloudlink.message.handler.MessageHandler.HandlerType;
import com.winit.cloudlink.message.internal.listener.ListenerContainer;
import com.winit.cloudlink.message.internal.rabbitmq.RabbitmqListenerContainer;
import com.winit.cloudlink.message.internal.rabbitmq.RabbitmqMessageTemplate;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.amqp.rabbit.connection.ConnectionFactory;

import java.util.Collection;
import java.util.Map;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.concurrent.locks.ReadWriteLock;
import java.util.concurrent.locks.ReentrantReadWriteLock;

/**
 * Created by stvli on 2015/11/10.
 */
public class DefaultMessageEngine implements MessageEngine {
	private static final Logger logger = LoggerFactory.getLogger(DefaultMessageEngine.class);
	private final Metadata metadata;
	private final RabbitListenerRegistry rabbitListenerRegistry = new RabbitListenerRegistry();
	private volatile MessageTemplate messageTemplate;
	private ReadWriteLock lock = new ReentrantReadWriteLock();
	private AtomicBoolean destroyed = new AtomicBoolean(false);
	private AtomicBoolean actived = new AtomicBoolean(false);

	public DefaultMessageEngine(Metadata metadata) {
		this.metadata = metadata;
	}

	@Override
	public void send(Message message) {
		messageTemplate.send(message);
	}

	@Override
	public void register(MessageHandler<? extends Message> messageHandler, HandlerRegisterCallback callback) {
		if (messageHandler == null) {
			return;
		}
		try {
			lock.readLock().lock();
			RabbitmqListenerContainer container = new RabbitmqListenerContainer(metadata, this, messageHandler, messageTemplate.getConnectionFactory());
			if (this.metadata.getApplicationOptions().isAutoStartContainer()) {
				container.start();
			}
			rabbitListenerRegistry.addListenerContainer(messageHandler, container);
			if (callback != null) {
				callback.onCompleted(HandlerType.MESSAGE, messageHandler.getMessageType());
			}
		} finally {
			lock.readLock().unlock();
		}
	}

	@Override
	public void unregister(MessageHandler<? extends Message> messageHandler) {
		if (messageHandler == null) {
			return;
		}
		try {
			lock.readLock().lock();
			ListenerContainer container = rabbitListenerRegistry.removeListenerContainer(messageHandler);
			if (null != container) {
				container.shutdown();
			}
		} finally {
			lock.readLock().unlock();
		}
	}

	@Override
	public void registerMonitor(MessageMonitor messageMonitor, ExchangeType messageTag) {
		if (messageMonitor == null) {
			return;
		}
	}

	@Override
	public void registerMessageReturnedListener(MessageReturnedListener listener) {
		messageTemplate.setMessageReturnedListener(listener);
	}

	@Override
	public MessageBuilder newMessageBuilder() {
		return new MessageBuilder(metadata);
	}


	@Override
	public void start() {
		active();
	}

	@Override
	public void active() {
		if (!actived.get()) {
			try {
				lock.writeLock().lock();
				messageTemplate = new RabbitmqMessageTemplate(metadata);
				messageTemplate.active();
				//rabbitListenerRegistry.shutdown();
				if (this.metadata.getApplicationOptions().isAutoStartContainer()) {
					//rabbitListenerRegistry.resetConnectionFactory(messageTemplate.getConnectionFactory());
					rabbitListenerRegistry.start();
				}
				actived.set(true);
			} catch (Throwable ex) {
				throw new CloudlinkException("Active message-engine error.", ex);
			} finally {
				lock.writeLock().unlock();
			}
		}
	}

	@Override
	public void deactive() {
		if (actived.get()) {
			try {
				lock.writeLock().lock();
				rabbitListenerRegistry.shutdown();
				actived.set(false);
			} catch (Throwable ex) {
				throw new CloudlinkException("Deactive message-engine error.", ex);
			} finally {
				lock.writeLock().unlock();
			}
		}
	}

	@Override
	public void shutdown() {
		if (!destroyed.getAndSet(true)) {
			deactive();
			try {
				lock.writeLock().lock();
				messageTemplate.shutdown();
				rabbitListenerRegistry.clear();
			} finally {
				lock.writeLock().unlock();
			}

		}
	}

	public boolean isActived() {
		return actived.get();
	}

	private class RabbitListenerRegistry implements Lifecycle, Clearable {
		private final Map<String, ListenerContainer> listenerContainers = Maps.newConcurrentMap();

		protected void addListenerContainer(MessageHandler messageHandler, ListenerContainer listenerContainer) {
			listenerContainers.put(messageHandler.toString(), listenerContainer);
		}

		public ListenerContainer removeListenerContainer(MessageHandler messageHandler) {
			return listenerContainers.remove(messageHandler.toString());
		}

		public void resetConnectionFactory(ConnectionFactory connectionFactory) {
			Collection<ListenerContainer> containers = listenerContainers.values();
			for (ListenerContainer container : containers) {
				if (container instanceof RabbitmqListenerContainer) {
					((RabbitmqListenerContainer) container).resetConnectionFactory(connectionFactory);
				}
			}
		}

		public void start() {
			Collection<ListenerContainer> containers = listenerContainers.values();
			for (ListenerContainer container : containers) {
				container.start();
			}
		}

		@Override
		public void shutdown() {
			Collection<ListenerContainer> containers = listenerContainers.values();
			for (ListenerContainer container : containers) {
				container.shutdown();
			}
		}

		@Override
		public void clear() {
			listenerContainers.clear();
		}
	}

}
